package com.thwcompany.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * Created by IntelliJ IDEA2021.3
 *
 * @Author: Tenghw
 * @Date: 2022/10/23  20:16
 * @Description: 交换机路由消息给消息队列
 */
@Slf4j
@Service
@SuppressWarnings("all")
public class ReturnConfirmMessageService implements RabbitTemplate.ReturnCallback{

    @Autowired
    private RabbitTemplate rabbitTemplate;

    //需要给ReturnCallback赋值 不然不会走回调方法，默认是null
    @PostConstruct
    public void init(){
        //rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(this);
    }

    public void sendMessage0(String exchange, String routingkey, Object message) {
        CorrelationData correlation = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.convertAndSend(exchange, routingkey, message,correlation);
        log.info("exchange交换机= {} ,路由键= {} ，要路由的消息message:{} ,correlation.getId()= {}",exchange,routingkey, message,correlation.getId());
    }
    public void sendMessage(String exchange, String routingkey, Object message) {
        String messageId=UUID.randomUUID().toString();
        //构建消息体
        Message  msg= MessageBuilder.withBody(String.valueOf(message).getBytes())
                .setMessageId(messageId)// 设置消息唯一标识
                .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)//消息类型
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)//设置消息持久化
                .setContentEncoding("UTF-8")//消息编码
                .setExpiration("10000")//这条消息的过期时间，单位毫秒，过期后消息会被删除
                .setPriority(5)//优先级
                .setUserId("root")
                .setCorrelationId(messageId)
                .build();
        CorrelationData correlationData = new CorrelationData(messageId);
        //rabbitTemplate.convertAndSend(exchange, routingkey, msg);
        rabbitTemplate.convertAndSend(exchange, routingkey, msg,correlationData);
        log.info("exchange交换机= {} ,路由键= {} ，要路由的消息message:{} ,messageId= {}",exchange,routingkey, message.toString(),msg.getMessageProperties().getMessageId());
    }

    //处理交换机发送消息到消息队列失败，则执行此方法。
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        log.info("交换机路由消息到消息队列失败=====》");
        log.info("exchange交换机名 = {}",exchange);
        log.info("routingKey路由键 = {}",routingKey);
        log.info("message要路由的消息 = {}",new String(message.getBody(), StandardCharsets.UTF_8));
        log.info("replyCode返回code = {}",replyCode);
        log.info("replyText = {}",replyText);
        if(replyCode == 312) {
            // 新起一个异步线程，重新发送一遍消息
            CompletableFuture.runAsync(() -> {
                log.info("retry send message one more time when trigger ReturnCallback message");
                rabbitTemplate.convertAndSend(exchange, routingKey, message,new CorrelationData(message.getMessageProperties().getMessageId()));
            });
        }
}




}
